package com.homedo.async.mysql.support.mapper.select;

import com.google.common.collect.Lists;
import com.homedo.async.mysql.support.dml.DMLEnum;
import com.homedo.async.mysql.support.mapper.util.CreateFlux4RecordResultUtil;
import com.homedo.async.mysql.support.mapper.util.CreateMono4CountNumberUtil;
import com.homedo.async.mysql.support.mapper.util.CreateMono4RecordResultUtil;
import com.homedo.async.mysql.support.meta.EntityMetaData;
import com.homedo.async.mysql.support.prepare.*;
import com.homedo.plainly.logger.Logger;
import com.homedo.plainly.logger.LoggerFactory;
import io.vertx.sqlclient.Pool;
import io.vertx.sqlclient.SqlConnection;
import org.reactivestreams.Publisher;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * @author jgshun
 * @date 2020/3/17 10:30 AM
 */
public class SelectMapperImpl<T, ID> implements SelectMapperI<T, ID> {
    private static final Logger _LOGGER = LoggerFactory.getLogger(SelectMapperImpl.class);

    private Pool pool;

    private EntityMetaData<T> entityMetaData;

    public SelectMapperImpl(Pool pool, EntityMetaData<T> entityMetaData) {
        this.pool = pool;
        this.entityMetaData = entityMetaData;
    }


    @Override
    public Mono<T> findById(ID id) {
        Assert.notNull(id, "The given id must not be null!");

        Object[] prepareSqlArgs = new PrepareRecordId2Sql(entityMetaData, DMLEnum.SELECT, Lists.newArrayList(id)).prepare();

        return this.createSelectMono(prepareSqlArgs, null);
    }

    @Override
    public Mono<T> findById(Publisher<ID> publisher) {
        Assert.notNull(publisher, "The given id must not be null!");

        return Mono.from(publisher).flatMap((id) -> {
            return this.findById(id);
        });

//        Mono<Optional<T>> optionalMono = Mono.create(sink -> {
//
//            publisher.subscribe(new Subscriber<ID>() {
//                @Override
//                public void onSubscribe(Subscription s) {
//                    s.request(1);//限定一个请求
//                }
//
//                @Override
//                public void onNext(ID nextId) {
//                    Collector<Row, ?, List<T>> collector = entityMetaData.getCollector();
//
//                    Object[] prepareSqlArgs = new PrepareRecordId2Sql(entityMetaData, DMLEnum.SELECT, nextId).prepare();
//
//                    pool.preparedQuery(prepareSqlArgs[0].toString(), Tuple.tuple((List<Object>) prepareSqlArgs[1]), collector, res -> {
//                        if (res.succeeded()) {
//                            SqlResult<List<T>> result = res.result();
//                            if (result.size() > 0) {
//                                sink.success(Optional.of(result.value().get(0)));
//                            } else {
//                                sink.success(Optional.empty());
//                            }
//                        } else {
//                            sink.error(res.cause());
//                            _LOGGER.error("Failure: {}", res.cause().getMessage(), res.cause());
//                        }
//                    });
//                }
//
//                @Override
//                public void onError(Throwable t) {
//                    sink.error(t);
//                }
//
//                @Override
//                public void onComplete() {
//                    //暂时无用
//                    //TODO 如果 id 不传数据指定Complete完成的话，有可能会导致 optionalMono 对象不会被释放
//                }
//            });
//        });
//
//        return optionalMono;
    }

    @Override
    public Mono<Boolean> existsById(ID id) {
        Assert.notNull(id, "The given id must not be null!");

        Mono<Boolean> booleanMono = Mono.create(sink -> {
            this.findById(id).doOnSuccess(e -> {
                sink.success(e != null);
            }).doOnError(error -> sink.error(error)).subscribe();

        });

        return booleanMono;
    }

    @Override
    public Mono<Boolean> existsById(Publisher<ID> publisher) {

        Assert.notNull(publisher, "The given id must not be null!");

        return Mono.from(publisher).flatMap(id -> {
            return this.existsById(id);
        });
    }

    @Override
    public Flux<T> findAll() {

        Object[] prepareSqlArgs = new PrepareObject2Sql(null, entityMetaData, DMLEnum.SELECT, PrepareObjectType.ENTITY_RECORD, false).prepare();

        return this.createSelectFlux(prepareSqlArgs, null);

        //TODO 用游标会触发 The statement (1) has no open cursor. 问题，或许是mysql bug 导致的？？
//     Collector<Row, ?, List<T>> collector = this.entityMetaData.getCollector();
//
//        Object[] prepareSqlArgs = new PrepareObject2Sql(null, entityMetaData, DMLEnum.SELECT, PrepareObjectType.ENTITY_RECORD, false).prepare();
//
//        Flux<T> optionalFlux = Flux.create(sink -> {
//            this.pool.getConnection(connRes -> {
//                if (connRes.succeeded()) {
//                    SqlSqlConnection connection = connRes.result();
//
//                    connection.prepare(prepareSqlArgs[0].toString(), preHandler -> {
//                        if (preHandler.succeeded()) {
//                            PreparedQuery preparedQuery = preHandler.result();
//
//                            Cursor cursor = preparedQuery.cursor(Tuple.tuple((List<Object>) prepareSqlArgs[1]));
//                            readFromCursor(cursor, collector, sink);
//
//                            //用流代替光标操作
////                            RowStream<Row> stream = preparedQuery.createStream(3000, Tuple.tuple((List<Object>) prepareSqlArgs[1]));
////
////                            stream.exceptionHandler(err -> {
////                                sink.error(err);
////                                connection.close();
////                            });
////                            stream.endHandler(v -> {
////                                sink.complete();
////                                connection.close();
////                            });
////
////                            stream.handler(row -> {
////                                sink.next(Stream.of(row).collect(collector).get(0));
////                            });
//                        } else {
//                            sink.error(preHandler.cause());
//                            connection.close();
//                        }
//                    });
//                } else {
//                    sink.error(connRes.cause());
//                }
//            });
//
//        });
//
//        return optionalFlux;
    }

    @Override
    public Flux<T> findByRecordSelective(T entity) {
        return this.findByRecord(entity, false, null);
    }

    @Override
    public Flux<T> findByRecordSelective(Publisher<? extends T> entityStream) {
        return this.findByRecord(entityStream, false, null);
    }

    @Override
    public Flux<T> findByRecordsSelective(Iterable<? extends T> entities) {
        return this.findByRecords(entities, false, null);
    }

    @Override
    public Flux<T> findByRecordsSelective(Publisher<? extends T> entityStream) {
        return this.findByRecords(entityStream, false, null);
    }

    @Override
    public Flux<T> findByRecord(T entity) {
        return this.findByRecord(entity, true, null);
    }

    @Override
    public Flux<T> findByRecord(Publisher<? extends T> entityStream) {
        return this.findByRecord(entityStream, true, null);
    }

    @Override
    public Flux<T> findByRecords(Iterable<? extends T> entities) {
        return this.findByRecords(entities, true, null);
    }

    @Override
    public Flux<T> findByRecords(Publisher<? extends T> entityStream) {
        return this.findByRecords(entityStream, true, null);

//        Flux<Optional<List<T>>> optionalFlux = Flux.create(sink -> {
//
//            entityStream.subscribe(new Subscriber<T>() {
//                @Override
//                public void onSubscribe(Subscription s) {
//                    s.request(1);//限定一个请求
//                }
//
//                @Override
//                public void onNext(T nextRecord) {
//                    findByRecord(nextRecord).subscribe(res -> {
//                        sink.next(res);
//                        sink.complete();
//                    }, error -> {
//                        sink.error(error);
//                    });
//                }
//
//                @Override
//                public void onError(Throwable t) {
//                    sink.error(t);
//                }
//
//                @Override
//                public void onComplete() {
//                    //暂时无用
//                    //TODO 如果 entityStream 不传数据就指定Complete完成的话，有可能会导致 optionalFlux 对象不会被释放
//                }
//            });
//
//        });
//
//        return optionalFlux;
    }

    @Override
    public Flux<T> findByIds(Iterable<ID> ids) {
        Assert.notNull(ids, "The given Iterable of Id's must not be null!");

        Object[] prepareSqlArgs = new PrepareRecordId2Sql(entityMetaData, DMLEnum.SELECT, ids).prepare();

        return this.createSelectFlux(prepareSqlArgs, null);
    }

    @Override
    public Flux<T> findByIds(Publisher<ID> idStream) {
        Assert.notNull(idStream, "The given Publisher of Id's must not be null!");

        return Flux.from(idStream).buffer().flatMap(this::findByIds);
    }

    @Override
    public Mono<Long> countAll() {
        return this.countByRecordSelective((T) null);
    }

    @Override
    public Mono<Long> countByRecordSelective(T entity) {
        return this.countByRecord(entity, false, null);
    }

    @Override
    public Mono<Long> countByRecordSelective(Publisher<? extends T> publisher) {
        return this.countByRecord(publisher, false, null);
    }

    @Override
    public Mono<Long> countByRecord(T entity) {
        return this.countByRecord(entity, true, null);
    }

    @Override
    public Mono<Long> countByRecord(Publisher<? extends T> publisher) {
        return this.countByRecord(publisher, true, null);
    }

    @Override
    public Mono<Long> countByRecordsSelective(Iterable<? extends T> entities) {
        return this.countByRecords(entities, false, null);
    }

    @Override
    public Mono<Long> countByRecordsSelective(Publisher<? extends T> entityStream) {
        return this.countByRecords(entityStream, false, null);
    }

    @Override
    public Mono<Long> countByRecords(Iterable<? extends T> entities) {
        return this.countByRecords(entities, true, null);
    }

    @Override
    public Mono<Long> countByRecords(Publisher<? extends T> entityStream) {
        return this.countByRecords(entityStream, true, null);
    }


    //==============================

    @Override
    public Mono<T> findById(ID id, SqlConnection connection) {
        Assert.notNull(id, "The given id must not be null!");

        Object[] prepareSqlArgs = new PrepareRecordId2Sql(entityMetaData, DMLEnum.SELECT, Lists.newArrayList(id)).prepare();

        return this.createSelectMono(prepareSqlArgs, connection);
    }

    @Override
    public Mono<T> findById(Publisher<ID> publisher, SqlConnection connection) {
        Assert.notNull(publisher, "The given id must not be null!");

        return Mono.from(publisher).flatMap((id) -> {
            return this.findById(id, connection);
        });
    }

    @Override
    public Mono<Boolean> existsById(ID id, SqlConnection connection) {
        Assert.notNull(id, "The given id must not be null!");

        Mono<Boolean> booleanMono = Mono.create(sink -> {
            this.findById(id, connection).doOnSuccess(e -> {
                sink.success(e != null);
            }).doOnError(error -> sink.error(error)).subscribe();

        });

        return booleanMono;
    }

    @Override
    public Mono<Boolean> existsById(Publisher<ID> publisher, SqlConnection connection) {

        Assert.notNull(publisher, "The given id must not be null!");

        return Mono.from(publisher).flatMap(id -> {
            return this.existsById(id, connection);
        });
    }

    @Override
    public Flux<T> findAll(SqlConnection connection) {

        Object[] prepareSqlArgs = new PrepareObject2Sql(null, entityMetaData, DMLEnum.SELECT, PrepareObjectType.ENTITY_RECORD, false).prepare();

        return this.createSelectFlux(prepareSqlArgs, connection);
    }

    @Override
    public Flux<T> findByRecordSelective(T entity, SqlConnection connection) {
        return this.findByRecord(entity, false, connection);
    }

    @Override
    public Flux<T> findByRecordSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.findByRecord(entityStream, false, connection);
    }

    @Override
    public Flux<T> findByRecordsSelective(Iterable<? extends T> entities, SqlConnection connection) {
        return this.findByRecords(entities, false, connection);
    }

    @Override
    public Flux<T> findByRecordsSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.findByRecords(entityStream, false, connection);
    }

    @Override
    public Flux<T> findByRecord(T entity, SqlConnection connection) {
        return this.findByRecord(entity, true, connection);
    }

    @Override
    public Flux<T> findByRecord(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.findByRecord(entityStream, true, connection);
    }

    @Override
    public Flux<T> findByRecords(Iterable<? extends T> entities, SqlConnection connection) {
        return this.findByRecords(entities, true, connection);
    }

    @Override
    public Flux<T> findByRecords(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.findByRecords(entityStream, true, connection);
    }

    @Override
    public Flux<T> findByIds(Iterable<ID> ids, SqlConnection connection) {
        Assert.notNull(ids, "The given Iterable of Id's must not be null!");

        Object[] prepareSqlArgs = new PrepareRecordId2Sql(entityMetaData, DMLEnum.SELECT, ids).prepare();

        return this.createSelectFlux(prepareSqlArgs, connection);
    }

    @Override
    public Flux<T> findByIds(Publisher<ID> idStream, SqlConnection connection) {
        Assert.notNull(idStream, "The given Publisher of Id's must not be null!");

        return Flux.from(idStream).buffer().flatMap(e -> {
            return this.findByIds(e, connection);
        });
    }

    @Override
    public Mono<Long> countAll(SqlConnection connection) {
        return this.countByRecordSelective((T) null, connection);
    }

    @Override
    public Mono<Long> countByRecordSelective(T entity, SqlConnection connection) {
        return this.countByRecord(entity, false, connection);
    }

    @Override
    public Mono<Long> countByRecordSelective(Publisher<? extends T> publisher, SqlConnection connection) {
        return this.countByRecord(publisher, false, connection);
    }

    @Override
    public Mono<Long> countByRecord(T entity, SqlConnection connection) {
        return this.countByRecord(entity, true, connection);
    }

    @Override
    public Mono<Long> countByRecord(Publisher<? extends T> publisher, SqlConnection connection) {
        return this.countByRecord(publisher, true, connection);
    }

    @Override
    public Mono<Long> countByRecordsSelective(Iterable<? extends T> entities, SqlConnection connection) {
        return this.countByRecords(entities, false, connection);
    }

    @Override
    public Mono<Long> countByRecordsSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.countByRecords(entityStream, false, connection);
    }

    @Override
    public Mono<Long> countByRecords(Iterable<? extends T> entities, SqlConnection connection) {
        return this.countByRecords(entities, true, connection);
    }

    @Override
    public Mono<Long> countByRecords(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.countByRecords(entityStream, true, connection);
    }

//    private void readFromCursor(Cursor cursor, Collector<Row, ?, List<T>> collector, FluxSink<T> sink) {
//        cursor.read(3000, rowRes -> {
//            if (rowRes.succeeded()) {
//                RowSet<Row> rows = rowRes.result();
//
//                List<T> entityRes = StreamSupport.stream(rows.spliterator(), false).collect(collector);
//
//                entityRes.forEach(e -> sink.next(e));
//
//                if (cursor.hasMore()) {
//                    readFromCursor(cursor, collector, sink);
//                } else {
//                    sink.complete();
//                    cursor.close();
//                }
//            } else {
//                sink.error(rowRes.cause());
//            }
//        });
//    }

    private Flux<T> findByRecord(T entity, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(entity, "The given entity must not be null!");

        Object[] prepareSqlArgs = new PrepareObject2Sql(entity, entityMetaData, DMLEnum.SELECT, PrepareObjectType.ENTITY_RECORD, includeNullVals).prepare();

        return this.createSelectFlux(prepareSqlArgs, connection);
    }

    private Flux<T> findByRecord(Publisher<? extends T> entityStream, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(entityStream, "The given Publisher of entity must not be null!");

        return Flux.create(sink -> {
            Mono.from(entityStream).subscribe(e -> {
                this.findByRecord(e, includeNullVals, connection).subscribe(res -> {
                    sink.next(res);
                }, error -> sink.error(error), new Runnable() {
                    @Override
                    public void run() {
                        sink.complete();
                    }
                });
            }, error -> sink.error(error));
        });
    }

    private Flux<T> findByRecords(Iterable<? extends T> entities, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(entities, "The given Iterable of entity's must not be null!");

        Object[] prepareSqlArgs = new PrepareRecords2Sql(entityMetaData, DMLEnum.SELECT, entities, includeNullVals).prepare();

        return this.createSelectFlux(prepareSqlArgs, connection);
    }

    private Flux<T> findByRecords(Publisher<? extends T> entityStream, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(entityStream, "The given Publisher of entity's must not be null!");

        return Flux.from(entityStream).buffer().flatMap(e -> {
            return this.findByRecords(e, includeNullVals, connection);
        });
    }

    private Mono<Long> countByRecord(T entity, boolean includeNullVals, SqlConnection connection) {
        Object[] prepareSqlArgs = new PrepareRecordCount2Sql(entity, entityMetaData, includeNullVals).prepare();

        return this.createCountMono(prepareSqlArgs, connection);
    }

    private Mono<Long> countByRecord(Publisher<? extends T> publisher, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(publisher, "The given publisher must not be null!");

        return Mono.from(publisher).flatMap((record) -> {
            return this.countByRecord(record, includeNullVals, connection);
        });
    }

    private Mono<Long> countByRecords(Iterable<? extends T> entities, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(entities, "The given Iterable of entity's must not be null!");

        Object[] prepareSqlArgs = new PrepareRecordsCount2Sql(entityMetaData, entities, includeNullVals).prepare();

        return this.createCountMono(prepareSqlArgs, connection);
    }

    private Mono<Long> countByRecords(Publisher<? extends T> entityStream, boolean includeNullVals, SqlConnection connection) {
        Assert.notNull(entityStream, "The given Publisher of entity's must not be null!");

        return Mono.create(sink -> {
            Flux.from(entityStream).buffer().subscribe(e -> {
                this.countByRecords(e, includeNullVals, connection).subscribe(res -> {
                    sink.success(res);
                }, error -> sink.error(error));
            }, error -> sink.error(error));
        });
    }

    private Mono<T> createSelectMono(Object[] prepareSqlArgs, SqlConnection connection) {
        return CreateMono4RecordResultUtil.create(prepareSqlArgs, this.pool, this.entityMetaData.getCollector(), connection);
    }

    private Flux<T> createSelectFlux(Object[] prepareSqlArgs, SqlConnection connection) {
        return CreateFlux4RecordResultUtil.create(prepareSqlArgs, this.pool, this.entityMetaData.getCollector(), connection);
    }

    private Mono<Long> createCountMono(Object[] prepareSqlArgs, SqlConnection connection) {
        return CreateMono4CountNumberUtil.create(prepareSqlArgs, this.pool, connection);
    }

}
